package com.jddup.program.dtm.postgresql;

import com.jddup.common.pojo.TableColumn;
import com.jddup.common.pojo.TableMeta;
import com.jddup.context.JddConfigContext;
import com.jddup.program.dtm.DbToTableMetaConvertor;
import org.apache.commons.lang3.StringUtils;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by ZZB on 2017/12/4.
 */
public class PostgresqlDbToTableMetaConvertor implements DbToTableMetaConvertor {
    private static final String PostgresqlTableMetaSQL =
            "SELECT\n" +
                    "  column_name as columnName,\n" +
                    "  upper(udt_name) as columnType,\n" +
                    "  character_maximum_length as maxLength,\n" +
                    "  numeric_precision as precision,\n" +
                    "  numeric_scale as scale,\n" +
                    "  is_nullable as isNullable,\n" +
                    "  column_default as defaultValue\n" +
                    "FROM information_schema.\"columns\"\n" +
                    "WHERE \"table_name\" = lower('{tableName}')\n" +
                    "      and \"table_schema\" = lower('{schemaName}')\n" +
                    "ORDER BY ordinal_position ASC;\n";

    private static final String PostgresqlTableMetaPrimarySQL =
            "SELECT pg_attribute.attname AS colname ,pg_constraint.conname as conname\n" +
                    "FROM pg_constraint\n" +
                    "  INNER JOIN pg_class\n" +
                    "    ON pg_constraint.conrelid = pg_class.oid\n" +
                    "  INNER JOIN pg_attribute ON pg_attribute.attrelid = pg_class.oid\n" +
                    "                             AND pg_attribute.attnum = pg_constraint.conkey [1]\n" +
                    "WHERE pg_class.relname = '{tableName}'\n" +
                    "      AND pg_constraint.contype = 'p';\n";

    @Override
    public TableMeta convertFromTableName(String tableFullName, String datasourceAlias) throws SQLException {
        TableMeta tableMeta = new TableMeta();
        tableMeta.setTableName(tableFullName);
        List<TableColumn> list = new ArrayList<>();

        Connection connection = JddConfigContext.getDataSourceMap().get(datasourceAlias).getConnection();
        if( !tableExist(connection,tableFullName))return null;
        String[] tableNameSplits = tableFullName.split("\\.");
        String schema = null;
        String tableName = null;
        if (tableNameSplits.length == 1) {
            schema = "public";
            tableName = tableNameSplits[0];
        } else {
            schema = tableNameSplits[0];
            tableName = tableNameSplits[1];
        }
        String pkColumnName = getPrimaryKeyColumn(tableName, connection);
        String sql = PostgresqlTableMetaSQL.replace("{tableName}", tableName)
                .replace("{schemaName}", schema);
        PreparedStatement ps = connection.prepareStatement(sql);
        ResultSet rs = ps.executeQuery();
        while (rs.next()) {
            TableColumn tableColumn = new TableColumn();
            tableColumn.setColumnName(rs.getString("columnName"));
            tableColumn.setColumnType(rs.getString("columnType"));
            tableColumn.setMaxLength(rs.getLong("maxLength"));
            tableColumn.setPrecision(rs.getLong("precision"));
            tableColumn.setScale(rs.getLong("scale"));
            tableColumn.setNullable(rs.getBoolean("isNullable"));
            tableColumn.setSigned(false);
            tableColumn.setDefaultValue(rs.getString("defaultValue"));
            tableColumn.setPrimaryKey(false);
            if (tableColumn.getColumnType().equals("BPCHAR")) {
                tableColumn.setColumnType("CHAR");
            }
            if (tableColumn.getDefaultValue() != null) {
                if (tableColumn.getDefaultValue().toLowerCase().startsWith("nextval('")
                        && tableColumn.getDefaultValue().toLowerCase().endsWith("serial_seq'::regclass)")) {
                    tableColumn.setDefaultValue("");
                }
            }
            tableColumn.setTypeFullName(buildPostgresFullTypeName(tableColumn.getColumnType(), tableColumn.getMaxLength(),
                    tableColumn.getPrecision(), tableColumn.getScale()));
            list.add(tableColumn);

            if (StringUtils.equals(StringUtils.lowerCase(tableColumn.getColumnName()), StringUtils.lowerCase(pkColumnName))) {
                tableColumn.setPrimaryKey(true);
                tableMeta.setPrimaryKeyColumn(tableColumn);
            }
        }
        tableMeta.setTableColumns(list);
        connection.close();
        return tableMeta;
    }

    @Override
    public boolean tableExist(Connection connection, String tableName) {
        try {
            Statement statement = connection.createStatement();
            ResultSet resultSet = statement.executeQuery("select 1 from \"" + tableName + "\" limit 1");
            resultSet.next();
            resultSet.close();
            statement.close();
            return true;
        } catch (SQLException e) {
            return false;
        }
    }

    private String getPrimaryKeyColumn(String tableName, Connection connection) throws SQLException {
        String primarySQL = PostgresqlTableMetaPrimarySQL.replace("{tableName}", tableName);
        Statement statement = connection.createStatement();
        ResultSet resultSet = statement.executeQuery(primarySQL);
        resultSet.next();
        String colname = resultSet.getString("colname");
        resultSet.close();
        statement.close();
        return colname;
    }

    private static String buildPostgresFullTypeName(
            String typeName, Long maxLength, Long precision, Long scale) {
        if ("CHAR".equals(typeName.toUpperCase()) || "VARCHAR".equals(typeName.toUpperCase()))
            if (maxLength != null && maxLength != 0) {
                typeName = typeName + "(" + maxLength + ")";
            }
        if ("NUMERIC".equals(typeName.toUpperCase()))
            if (precision != null && precision != 0) {
                if (scale != null && scale != 0) {
                    typeName = typeName + "(" + precision + "," + scale + ")";
                } else {
                    typeName = typeName + "(" + precision + ")";
                }
            }


        return typeName;
    }

}
